package main

import (
	"DMSKafka/config"
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/pkg/errors"
	"log"
	"time"
)

var delayMsg = []string{"start", "work", "stop", "close"}
var delayTopic = "testlog"

func main() {
	conf, err := config.GetConfig()
	if err != nil {
		log.Printf("%+v\n", err)
		return
	}
	//初始化同步生产者
	producerConfig := sarama.NewConfig()
	producerConfig.Producer.Return.Successes = true
	producerConfig.Producer.RequiredAcks = sarama.WaitForAll
	producer, err := sarama.NewSyncProducer(conf.Host, producerConfig)
	if err != nil {
		log.Printf("%+v\n", errors.Wrap(err, "初始化生产者失败"))
		return
	}
	defer producer.Close()
	for _, mes := range delayMsg {
		//构建消息体
		msg := &sarama.ProducerMessage{
			Topic:     delayTopic,
			Value:     sarama.ByteEncoder(mes),
			Timestamp: time.Now(),
		}
		//发送消息
		partition, offset, err := producer.SendMessage(msg)
		if err != nil {
			log.Printf("%+v\n", errors.Wrap(err, "发送消息失败"))
		}
		fmt.Printf("the partition %d,offset %d\n", partition, offset)
	}
}
